package com.talent.service.iot.mqtt.handler;

import com.talent.service.iot.mqtt.config.Constant;
import com.talent.service.iot.mqtt.model.entity.TlSubClient;
import com.talent.service.iot.mqtt.model.entity.TlTopic;
import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import com.talent.service.iot.mqtt.model.request.TlMqttSubscribeReq;
import com.talent.service.iot.mqtt.model.response.TlMqttSubAck;
import com.talent.service.iot.mqtt.service.IMqttRetainMsgService;
import com.talent.service.iot.mqtt.service.IMqttSessionService;
import com.talent.service.iot.mqtt.service.IMqttSubscriptionService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;

import java.util.List;
import java.util.Set;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
public class TlMqttSubScribeHandler extends SimpleChannelInboundHandler<TlMqttSubscribeReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttSubScribeHandler.class);

    private final IMqttSubscriptionService subscriptionService;

    private final IMqttSessionService sessionService;

    private final IMqttRetainMsgService retainMsgService;


    public TlMqttSubScribeHandler(IMqttSubscriptionService subscriptionService,IMqttSessionService sessionService,IMqttRetainMsgService retainMsgService){
        this.subscriptionService = subscriptionService;
        this.sessionService = sessionService;
        this.retainMsgService = retainMsgService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttSubscribeReq msg) throws Exception {
        logger.debug("in【SUBSCRIBE】handler");
        Channel channel = ctx.channel();
        List<TlTopic> topics =  msg.getPayload().getTopics();
        String clientId =channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get().toString();

         sessionService.find(clientId)
                 .subscribe(session -> {
            Set<TlTopic> subTopics = session.getTopics();
            subTopics.addAll(topics);
        });
        /**
         * 需要将客户端订阅的主题进行保存起来
         * 后面根据这个主题就能够立马找到对应的客户端 key就是主题的名称 最好能够获取到这个主题的订阅的qos等级,然后获取到这个客户端通道 将消息转发给他
         */
        Flux.fromIterable(topics)
                .subscribe(topic->{
                    TlSubClient client = new TlSubClient(topic.getQos(),clientId,topic.getName());
                    logger.debug("client【{}】subscribe topic 【{}】",clientId,topic.getName());
                    subscriptionService.subscribe(client);
                    //查询这个主题是否存在保留消息 如果有就发送
                    Flux<TlMqttPublishReq> retainMessage = retainMsgService.find(topic.getName());
                    retainMessage.subscribe(channel::writeAndFlush);
                });

        int messageId = msg.getVariableHead().getMessageId();
        TlMqttSubAck res = TlMqttSubAck.of(topics, messageId);
        channel.writeAndFlush(res);
        logger.debug("build【SUBACK】return");

    }
}
